package xin.marcher.module.inventory.mq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import xin.marcher.module.common.constants.RocketMqConstant;
import xin.marcher.module.common.exception.CommonErrorCodeEnum;
import xin.marcher.module.common.exception.ProductBizException;
import xin.marcher.module.common.utils.HashUtil;
import xin.marcher.module.inventory.mq.config.RocketMQProperties;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 *
 */
@Slf4j
@Component
public class DefaultProducer {

    private DefaultMQProducer producer;

    @Autowired
    public DefaultProducer(RocketMQProperties rocketMQProperties) {
        producer = new DefaultMQProducer(RocketMqConstant.ORDER_DEFAULT_PRODUCER_GROUP);
        producer.setNamesrvAddr(rocketMQProperties.getNameServer());
        start();
    }

    public DefaultMQProducer getProducer() {
        return this.producer;
    }

    /**
     * 对象在使用之前必须要调用一次，只能初始化一次
     */
    public void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            log.error("producer start error", e);
        }
    }


    /**
     * 一般在应用上下文，使用上下文监听器，进行关闭
     */
    public void shutdown() {
        this.producer.shutdown();
    }

    /**
     * 发送消息
     *
     * @param topic   topic
     * @param message 消息
     */
    public void sendMessage(String topic, String message, String type) {
        sendMessage(topic, message, -1, type);
    }

    /**
     * 发送异步消息
     *
     * @param topic   topic
     * @param message 消息
     */
    public void sendAsyncMessage(String topic, String message, String type) {
        Message msg = new Message(topic, message.getBytes(StandardCharsets.UTF_8));
        try {
            //2、异步发送
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {

                }

                @Override
                public void onException(Throwable e) {
                    // 这里对于失败的消息，要做重试处理
                    log.error("发送MQ消息失败, type:{}, message:{}", type, message, e);
                }
            });
        } catch (Exception e) {
            log.error("发送MQ消息失败, type:{}, message:{}", type, message, e);
            throw new ProductBizException(CommonErrorCodeEnum.SEND_MQ_FAILED);
        }
    }

    /**
     * 发送消息
     *
     * @param topic   topic
     * @param message 消息
     */
    public void sendMessage(String topic, String message, Integer delayTimeLevel, String type) {
        Message msg = new Message(topic, message.getBytes(StandardCharsets.UTF_8));
        try {
            if (delayTimeLevel > 0) {
                msg.setDelayTimeLevel(delayTimeLevel);
            }
            SendResult send = producer.send(msg);
            if (SendStatus.SEND_OK == send.getSendStatus()) {
                log.info("发送MQ消息成功, type:{}, message:{}", type, message);
            } else {
                throw new ProductBizException(send.getSendStatus().toString());
            }
        } catch (Exception e) {
            log.error("发送MQ消息失败, type:{}, message:{}", type, message, e);
            throw new ProductBizException(CommonErrorCodeEnum.SEND_MQ_FAILED);
        }
    }

    /**
     * 自定义 指定消息写入固定队列
     *
     * @param topic   消息topic
     * @param message 消息内容
     * @param key     消息路由key
     * @param type    消息描述
     * @throws Exception
     */
    public void sendMessage(String topic, String message, String key, String type) throws Exception {
        Message msg = new Message(topic, "taga", key, message.getBytes(StandardCharsets.UTF_8));
        try {
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    long hash = HashUtil.murMurHash(o.toString().getBytes());
                    int queueNum = (int) (hash %= list.size());
                    return list.get(queueNum);
                }
            }, 0);
            if (SendStatus.SEND_OK == sendResult.getSendStatus()) {
                log.info("发送MQ消息成功, type:{}, message:{}", type, message);
            } else {
                throw new ProductBizException(sendResult.getSendStatus().toString());
            }
        } catch (Exception e) {
            log.error("发送MQ消息失败, type:{}, message:{}", type, message, e);
            throw new ProductBizException(CommonErrorCodeEnum.SEND_MQ_FAILED);
        }
    }
}
